package com.danan.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.danan.realtime.app.BaseApp;
import com.danan.realtime.common.Constant;
import com.danan.realtime.util.SinkUtil;
import com.danan.realtime.util.TimeUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;

/**
 * @author NanHuang
 * @Date 2023/1/24
 */
public class DwdApp_1_traffic extends BaseApp {
    private static final String DS_START_KEY = "start";
    private static final String DS_PAGE_KEY = "key";
    private static final String DS_ERR_KEY = "err";
    private static final String DS_ACTION_KEY = "action";
    private static final String DS_DISPLAY_KEY = "display";

    public static void main(String[] args) {
        new DwdApp_1_traffic().init(4001,3,"DwdApp_1_traffic", Constant.TOPIC_ODS_LOG);
    }

    @Override
    public void invoke(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        // 1 对数据进行ETL，过滤脏数据
        SingleOutputStreamOperator<JSONObject> etledDs = etl(ds);
        // 2 纠正“is_new”字段数据
        SingleOutputStreamOperator<JSONObject> correctedDs = correctData(etledDs);
        // 3 数据分流
        HashMap<String, DataStream<JSONObject>> result = dataDiversion(correctedDs);
        // 4 将数据写入到Kafka中
        writeToKafka(result);
    }

    private void writeToKafka(HashMap<String, DataStream<JSONObject>> result) {
        result.get(DS_START_KEY).addSink(SinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        result.get(DS_PAGE_KEY).addSink(SinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        result.get(DS_ERR_KEY).addSink(SinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        result.get(DS_ACTION_KEY).addSink(SinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        result.get(DS_DISPLAY_KEY).addSink(SinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
    }

    private HashMap<String, DataStream<JSONObject>> dataDiversion(SingleOutputStreamOperator<JSONObject> ds) {
        OutputTag<JSONObject> pageOp = new OutputTag<>("page", TypeInformation.of(JSONObject.class));
        OutputTag<JSONObject> errOp = new OutputTag<>("err", TypeInformation.of(JSONObject.class));
        OutputTag<JSONObject> actionOp = new OutputTag<>("action", TypeInformation.of(JSONObject.class));
        OutputTag<JSONObject> displayOp = new OutputTag<>("display", TypeInformation.of(JSONObject.class));
        SingleOutputStreamOperator<JSONObject> outputStream = ds.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject in, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                JSONObject common = in.getJSONObject("common");
                Long ts = in.getLong("ts");
                // start
                JSONObject start = in.getJSONObject("start");
                if (start != null) {
                    start.putAll(common);
                    start.put("ts", ts);
                    out.collect(start);
                }
                // page
                JSONObject page = in.getJSONObject("page");
                if (page != null) {
                    page.putAll(common);
                    page.put("ts", ts);
                    ctx.output(pageOp, page);
                }
                // err
                JSONObject err = in.getJSONObject("err");
                if (err != null) {
                    err.putAll(common);
                    err.put("ts", ts);
                    ctx.output(errOp, err);
                }
                // action
                JSONArray actions = in.getJSONArray("actions");
                if (actions != null){
                    for (int i = 0; i < actions.size(); i++) {
                        JSONObject action = actions.getJSONObject(i);
                        action.putAll(common);
                        action.put("ts", ts);
                        ctx.output(actionOp, action);
                    }
                }
                // display
                JSONArray displays = in.getJSONArray("displays");
                if (displays != null){
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject display = displays.getJSONObject(i);
                        display.putAll(common);
                        display.put("ts", ts);
                        ctx.output(displayOp, display);
                    }
                }
            }
        });
        DataStream<JSONObject> pageDs = outputStream.getSideOutput(pageOp);
        DataStream<JSONObject> errDs = outputStream.getSideOutput(errOp);
        DataStream<JSONObject> actionDs = outputStream.getSideOutput(actionOp);
        DataStream<JSONObject> displayDs = outputStream.getSideOutput(displayOp);
        HashMap<String, DataStream<JSONObject>> result = new HashMap<>();
        result.put(DS_START_KEY,outputStream);
        result.put(DS_PAGE_KEY,pageDs);
        result.put(DS_ERR_KEY,errDs);
        result.put(DS_ACTION_KEY,actionDs);
        result.put(DS_DISPLAY_KEY,displayDs);
        return result;
    }

    private SingleOutputStreamOperator<JSONObject> correctData(SingleOutputStreamOperator<JSONObject> ds) {
        return ds.keyBy(j -> j.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<String> flagDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        flagDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("flagDate", String.class));
                    }

                    @Override
                    public void processElement(JSONObject in, Context ctx, Collector<JSONObject> out) throws Exception {
                        String dateStr = TimeUtil.parseDateStr(in.getLong("ts"), false);
                        JSONObject common = in.getJSONObject("common");
                        String flag = flagDate.value();
                        if ("1".equals(common.getString("is_new"))){
                            if (flag == null) {
                                flagDate.update(dateStr);
                            } else {
                                if (!flag.equals(dateStr)) {
                                    common.put("is_new","0");
                                }
                            }
                        } else {
                            if (flag == null) {
                                flagDate.update(TimeUtil.parseDateStr(in.getLong("ts") - 24 * 60 * 60 * 1000,false));
                            }
                        }
                        out.collect(in);
                    }
                });
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> ds) {
        return ds.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject obj = JSON.parseObject(s);
                    return true;
                } catch (Exception e) {
                    System.out.println("数据格式错误，无法解析！（" + s + "）");
                    return false;
                }
            }
        }).map(JSON::parseObject);
    }
}
